package com.tx.manager;

import cn.hutool.core.collection.ConcurrentHashSet;
import com.alibaba.fastjson.JSONObject;
import com.tx.constant.WsEvent;
import com.tx.wsclient.WsClient;
import com.tx.wsclient.WsClientManager;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.listener.ChannelTopic;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author dengguoxing
 * @date 2019/11/7
 * description
 **/
public class BookTopicsManager {

    private static final Logger logger = LoggerFactory.getLogger(BookTopicsManager.class);

    /**
     * topic - set[id, ...]
     */
    public static Map<String, Set<String>> topicsCache = new HashMap<>(256);

    public static Set<String> addTopics = new ConcurrentHashSet<>(4);

    public static Set<String> delTopics = new ConcurrentHashSet<>(4);


    /**
     * 用来标识订阅的主题是否有修改过，用于更新redis订阅。
     * 需要保证线程安全
     */
    public static boolean updateFlag = false;

    /**
     * 客户端新增订阅
     *
     * @param id
     * @param topic
     */
    public static void subTopic(String id, String topic) {
        synchronized (topicsCache) {
            Set<String> idSet = topicsCache.get(topic);
            if (idSet == null) {
                idSet = new HashSet(4);
                idSet.add(id);
                topicsCache.put(topic, idSet);
                changeUpdateFlag(true);
                boolean isSuccess = delTopics.remove(topic);
                if (!isSuccess) {
                    addTopics.add(topic);
                    System.out.println(addTopics);
                }
            } else {
                idSet.add(id);
            }
        }
        logger.info("assetTopicsManager add topic, id: 【{}】, topic: 【{}】", id, topic);
    }

    /**
     * 客户端取消订阅
     *
     * @param id
     * @param topic
     */
    public static void unSubTopic(String id, String topic) {
        synchronized (topicsCache) {
            Set<String> idSet = topicsCache.get(topic);
            if (idSet != null) {
                idSet.remove(id);
                if (idSet.size() == 0) {
                    topicsCache.remove(topic, idSet);
                    changeUpdateFlag(true);
                    boolean isSuccess = addTopics.remove(topic);
                    if (!isSuccess) {
                        delTopics.add(topic);
                    }
                }
            }
        }
        logger.info("assetTopicsManager remove topic, id: 【{}】, topic: 【{}】", id, topic);
    }

    /**
     * 获取订阅主题的变化
     * Pair<delTopics, addTopics>
     *
     * @return
     */
    public static Pair<Set<ChannelTopic>, Set<ChannelTopic>> getChannelTopics() {
        synchronized (topicsCache) {
            changeUpdateFlag(false);
            Set<ChannelTopic> left = null;
            Set<ChannelTopic> right = null;
            if (delTopics.size() > 0) {
                left = delTopics.stream().map(x -> new ChannelTopic(x)).collect(Collectors.toSet());
                delTopics.clear();
            }
            if (addTopics.size() > 0) {
                right = addTopics.stream().map(x -> new ChannelTopic(x)).collect(Collectors.toSet());
                addTopics.clear();
            }
            logger.info("检测订阅任务--asset订阅主题列表: {}", topicsCache.keySet());
            return Pair.of(left, right);
        }
    }


    public static boolean isUpdateFlag() {
        return updateFlag;
    }

    /**
     * 更新标签状态
     *
     * @param newFlag
     */
    public static synchronized void changeUpdateFlag(boolean newFlag) {
        updateFlag = newFlag;
    }

    /**
     * 推送信息到订阅的用户
     *
     * @param topic
     * @param data
     */
    public static void sendByTopic(String topic, JSONObject data) {
        Set<String> ids = topicsCache.get(topic);
        if (ids != null) {
            for (String id : ids) {
                WsClient wsClient = WsClientManager.getWsClient(id);
                if (wsClient != null) {
                    wsClient.send(WsEvent.PUSH.getValue(), topic, data);
                }
            }
        }
    }
}
